package com.wfg.flink.example.windows;

import com.wfg.flink.example.function.DataSplitter;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.*;
import org.apache.flink.streaming.api.windowing.time.Time;

import static com.wfg.flink.example.constants.Constants.TOPIC_NAME;

/**
 * Desc: Flink Window 学习
 */
@Slf4j
public class WindowsDemo {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String brokers = "localhost:9092";
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers(brokers)
                .setTopics(TOPIC_NAME)
                .setGroupId("my-group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStreamSource<String> data = env.fromSource(source, WatermarkStrategy.noWatermarks(), "wfgxxx");

        //基于时间窗口
//       data.flatMap(new DataSplitter())
//                .keyBy(1)
//                .timeWindow(Time.seconds(30))
//                .sum(0)
//                .print();

        //基于滑动时间窗口
/*        data.flatMap(new DataSplitter())
                .keyBy(1)
                .timeWindow(Time.seconds(60), Time.seconds(30))
                .sum(0)
                .print();*/


        //基于事件数量窗口
/*        data.flatMap(new DataSplitter())
                .keyBy(1)
                .countWindow(3)
                .sum(0)
                .print();*/


        //基于事件数量滑动窗口
/*        data.flatMap(new DataSplitter())
                .keyBy(1)
                .countWindow(4, 3)
                .sum(0)
                .print();*/


        //基于会话时间窗口
//        data.flatMap(new DataSplitter())
//                .keyBy(v->v.f0)
//                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
//                //表示如果 5s 内没出现数据则认为超出会话时长，然后计算这个窗口的和
//                .sum(1)
//                .print();

        //滚动窗口（Tumbling Window） 基于处理时间的 30 秒滚动窗口
//        data.flatMap(new DataSplitter())
//                .keyBy(v->v.f0)
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(30)))
//                .sum(1)
//                .print();
//
//        // 基于事件时间的 30 秒滚动窗口
//        data.flatMap(new DataSplitter())
//                .keyBy(v->v.f0)
//                .assignTimestampsAndWatermarks(/* 分配时间戳和水印 */)
//                .window(TumblingEventTimeWindows.of(Time.seconds(30)))
//                .sum(1)
//                .print();

        // 基于处理时间的 30 秒滑动窗口，滑动间隔为 10 秒
        data.flatMap(new DataSplitter())
                .keyBy(v->v.f0)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(30), Time.seconds(10)))
                .sum(1)
                .print();

        // 基于事件时间的 30 秒滑动窗口，滑动间隔为 10 秒
//        data.flatMap(new DataSplitter())
//                .keyBy(v->v.f0)
//                .assignTimestampsAndWatermarks(/* 分配时间戳和水印 */)
//                .window(SlidingEventTimeWindows.of(Time.seconds(30), Time.seconds(10)))
//                .sum(1)
//                .print();
        env.execute("flink window example");
    }
}
